package com.heiye.relation.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.heiye.framework.common.exception.BizException;
import com.heiye.framework.common.response.Response;
import com.heiye.framework.common.util.DateUtils;
import com.heiye.framework.common.util.JsonUtils;
import com.heiye.quiz.api.dto.resp.CheckQuizIsExistRespDTO;
import com.heiye.relation.api.dto.FindQuizFansTop3ByIdsReqDTO;
import com.heiye.relation.api.dto.FindQuizFansTop3ByIdsRspDTO;
import com.heiye.relation.biz.domain.dataobject.QuizFansDO;
import com.heiye.relation.biz.domain.mapper.QuizFansDOMapper;
import com.heiye.relation.biz.model.vo.FindQuizIsFollowReqVO;
import com.heiye.relation.biz.model.vo.FindQuizIsFollowRspVO;
import com.heiye.relation.biz.constant.MQConstants;
import com.heiye.relation.biz.constant.RedisKeyConstants;
import com.heiye.relation.biz.domain.dataobject.QuizFollowingDO;
import com.heiye.relation.biz.domain.mapper.QuizFollowingDOMapper;
import com.heiye.relation.biz.enums.LuaResultEnum;
import com.heiye.relation.biz.enums.ResponseCodeEnum;
import com.heiye.relation.biz.model.dto.FollowQuizMqDTO;
import com.heiye.relation.biz.model.dto.UnfollowQuizMqDTO;
import com.heiye.relation.biz.model.vo.FollowQuizReqVO;
import com.heiye.relation.biz.model.vo.UnfollowQuizReqVO;
import com.heiye.relation.biz.rpc.QuizRpcService;
import com.heiye.relation.biz.rpc.UserRpcService;
import com.heiye.relation.biz.service.QuizRelationService;
import com.heiye.user.api.dto.resp.FindUserByIdRspDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.heiye.framework.biz.context.thread.LoginUserContextHolder;
import org.springframework.core.io.ClassPathResource;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2025/04/15 上午11:26
 * @version: v1.0.0
 * @description: 问题关系业务
 */
@Service
@Slf4j
public class QuizRelationServiceImpl implements QuizRelationService {

    @Resource
    private QuizRpcService quizRpcService;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private QuizFollowingDOMapper quizFollowingDOMapper;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private QuizFansDOMapper quizFansDOMapper;

    @Resource
    private UserRpcService userRpcService;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 关注问题
     *
     * @param followQuizReqVO
     * @return
     */
    @Override
    public Response<?> follow(FollowQuizReqVO followQuizReqVO) {
        // 问题 ID
        Long followQuizId = followQuizReqVO.getFollowQuizId();

        // 获取线程中用户的ID
        Long userId = LoginUserContextHolder.getUserId();

        if (Objects.equals(userId, followQuizReqVO.getCreatorId())) {
            throw new BizException(ResponseCodeEnum.CANT_FOLLOW_YOUR_SELF);
        }

        // 判断问题是否存在
        CheckQuizIsExistRespDTO checkQuizIsExistRespDTO = quizRpcService.checkNoteIsExist(followQuizId);

        if (Objects.isNull(checkQuizIsExistRespDTO) || !checkQuizIsExistRespDTO.getIsExist()) {
            throw new BizException(ResponseCodeEnum.FOLLOW_QUIZ_NOT_EXISTED);
        }

        // 构建 Redis Key
        String quizFollowingRedisKey = RedisKeyConstants.buildRBitmapQuizFollowingKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_follow_check_and_add.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), followQuizId);
        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);
        if (Objects.isNull(luaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        // 判断返回结果
        switch (luaResultEnum) {
            // 已经关注了该问题
            case ALREADY_FOLLOWED -> throw new BizException(ResponseCodeEnum.ALREADY_FOLLOWED);
            // R.SETBIT 关注列表不存在
            case R_BITMAP_NOT_EXIST -> {

                // 随机过期时间
                // 保留 1 天 + 随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // 从数据库查询当前用户的关注关系记录
                List<QuizFollowingDO> quizFollowingDOS = quizFollowingDOMapper.selectByUserId(userId);

                // 若记录为空，直接 ZADD 关系数据，并设置过期时间
                if (CollUtil.isEmpty(quizFollowingDOS)) {

                    DefaultRedisScript<Long> script2 = new DefaultRedisScript<>();
                    // Lua 脚本路径
                    script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_follow_add_and_expire.lua")));
                    // 返回值类型
                    script2.setResultType(Long.class);

                    redisTemplate.execute(script2, Collections.singletonList(quizFollowingRedisKey), followQuizId, expireSeconds);
                }
                // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间
                else {
                    // 初始化 RBitmap
                    batchAddQuizFollowRBitmapAndExpire(quizFollowingDOS, expireSeconds, quizFollowingRedisKey);

                    // 再次调用上面的 Lua 脚本：rbitmap_follow_check_and_add.lua , 将最新的关注关系添加进去
                    result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), followQuizId);

                    checkLuaScriptResult(result);
                }
            }
        }

        // 发送 MQ
        // 构建消息体 DTO
        FollowQuizMqDTO followQuizMqDTO = FollowQuizMqDTO.builder()
                .userId(userId)
                .followQuizId(followQuizId)
                .createTime(LocalDateTime.now())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(followQuizMqDTO))
                .build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_FOLLOW;

        String hashKey = String.valueOf(userId);

        log.info("==> 开始发送关注操作 MQ, 消息体: {}", followQuizMqDTO);

        // 异步发送 MQ 消息，提升接口响应速度
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(destination, message, hashKey);

        log.info("==> MQ 发送结果，SendResult: {}", sendResult);

        return Response.success();
    }

    /**
     * 取关问题
     *
     * @param unfollowQuizReqVO
     * @return
     */
    @Override
    public Response<?> unfollow(UnfollowQuizReqVO unfollowQuizReqVO) {
        // 想要取关的问题 ID
        Long unfollowQuizId = unfollowQuizReqVO.getUnfollowQuizId();
        // 当前登录用户 ID
        Long userId = LoginUserContextHolder.getUserId();

        // 无法取关自己的问题
        if (Objects.equals(userId, unfollowQuizReqVO.getCreatorId())) {
            throw new BizException(ResponseCodeEnum.CANT_UNFOLLOW_YOUR_QUIZ_SELF);
        }

        // 校验关注的问题是否存在
        CheckQuizIsExistRespDTO checkQuizIsExistRespDTO = quizRpcService.checkNoteIsExist(unfollowQuizId);

        if (Objects.isNull(checkQuizIsExistRespDTO) || !checkQuizIsExistRespDTO.getIsExist()) {
            throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
        }

        // 必须是关注了的问题，才能取关
        String quizFollowingRedisKey = RedisKeyConstants.buildRBitmapQuizFollowingKey(userId);
        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_unfollow_check_and_delete.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), unfollowQuizId);
        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);

        if (Objects.isNull(luaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        // 校验 Lua 脚本执行结果
        // 取关的用户不在关注列表中
        switch (luaResultEnum) {
            case NOT_FOLLOWED -> throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
            case R_BITMAP_NOT_EXIST -> {
                // 从数据库查询当前用户的关注关系记录
                List<QuizFollowingDO> quizFollowingDOS = quizFollowingDOMapper.selectByUserId(userId);

                // 随机过期时间
                // 保底1天+随机秒数
                long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

                // 若记录为空，则表示还未关注任何人，提示还未关注对方
                if (CollUtil.isEmpty(quizFollowingDOS)) {
                    throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
                }
                // 若记录不为空，则将关注关系数据全量同步到 Redis 中，并设置过期时间
                else {
                    // 初始化 RBitmap
                    batchAddQuizFollowRBitmapAndExpire(quizFollowingDOS, expireSeconds, quizFollowingRedisKey);

                    // 再次调用上面的 Lua 脚本：rbitmap_unfollow_check_and_delete.lua , 将取关的用户删除
                    result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), unfollowQuizId);

                    if (Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode())) {
                        throw new BizException(ResponseCodeEnum.NOT_FOLLOWED);
                    }
                }


            }
        }

        // 发送 MQ
        // 构建消息体 DTO
        UnfollowQuizMqDTO unfollowQuizMqDTO = UnfollowQuizMqDTO.builder()
                .userId(userId)
                .unfollowQuizId(unfollowQuizId)
                .createTime(LocalDateTime.now())
                .build();

        // 构建消息对象，并将 DTO 转成 Json 字符串设置到消息体中
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(unfollowQuizMqDTO)).build();

        // 通过冒号连接, 可让 MQ 发送给主题 Topic 时，携带上标签 Tag
        String destination = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW + ":" + MQConstants.TAG_UNFOLLOW;

        String hashKey = String.valueOf(userId);

        log.info("==> 开始发送取关操作 MQ, 消息体: {}", unfollowQuizMqDTO);

        SendResult sendResult = rocketMQTemplate.syncSendOrderly(destination, message, hashKey);

        log.info("==> MQ 发送结果，SendResult: {}", sendResult);

        return Response.success();
    }

    /**
     * 查询问题是否关注
     *
     * @param findQuizIsFollowReqVO
     * @return
     */
    @Override
    public Response<FindQuizIsFollowRspVO> FindQuizIsFollow(FindQuizIsFollowReqVO findQuizIsFollowReqVO) {
        // 问题 ID
        Long quizId = findQuizIsFollowReqVO.getQuizId();

        // 获取登录用户 ID
        Long userId = LoginUserContextHolder.getUserId();

        // 构建 Redis Key
        String quizFollowingRedisKey = RedisKeyConstants.buildRBitmapQuizFollowingKey(userId);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // Lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_follow_exists_and_check.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        // 执行 Lua 脚本，拿到返回结果
        Long result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), quizId);

        // 构建返回结果, 根据未关注问题的枚举进行判断
        Boolean isFollow = !Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode());

        // result 不存在
        if (Objects.equals(LuaResultEnum.R_BITMAP_NOT_EXIST.getCode(), result)) {
            // 从数据库查询当前用户的关注关系记录
            List<QuizFollowingDO> quizFollowingDOS = quizFollowingDOMapper.selectByUserId(userId);

            // 随机过期时间
            // 保底1天+随机秒数
            long expireSeconds = 60 * 60 * 24 + RandomUtil.randomInt(60 * 60 * 24);

            // 用户从未关注过任何问题
            if (CollUtil.isEmpty(quizFollowingDOS)) {
                // 将一个完全不存在的问题问题将其置空初始化 rbitmap
                DefaultRedisScript<Long> script1 = new DefaultRedisScript<>();
                // lua 脚本路径
                script1.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_follow_add_and_expire.lua")));
                // 返回值类型
                script1.setResultType(Long.class);

                redisTemplate.execute(script1, Collections.singletonList(quizFollowingRedisKey), Long.MIN_VALUE, expireSeconds);

                isFollow = false;
            } else {
                // 初始化 RBitmap
                batchAddQuizFollowRBitmapAndExpire(quizFollowingDOS, expireSeconds, quizFollowingRedisKey);

                // 再次调用上面的 Lua 脚本：rbitmap_follow_exists_and_check.lua , 校验用户是否关注
                result = redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), quizId);

                isFollow = !Objects.equals(result, LuaResultEnum.NOT_FOLLOWED.getCode());
            }
        }

        // 构建返参
        FindQuizIsFollowRspVO findQuizIsFollowRspVO = FindQuizIsFollowRspVO.builder()
                .isFollow(isFollow)
                .build();

        return Response.success(findQuizIsFollowRspVO);
    }

    /**
     * 批量查询问题的前三个粉丝
     *
     * @param findQuizFansTop3ByIdsReqDTO
     * @return
     */
    @Override
    public Response<FindQuizFansTop3ByIdsRspDTO> findQuizFansTop3ByIds(FindQuizFansTop3ByIdsReqDTO findQuizFansTop3ByIdsReqDTO) {
        // 问题 IDS
        List<Long> quizIds = findQuizFansTop3ByIdsReqDTO.getQuizIds();

        // 用 redis pipelined 批量查询 提高效率
        List<Object> redisPipelineResult = redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public Object execute(RedisOperations operations) {
                for (Long quizId : quizIds) {
                    // 构建 Redis Key
                    String quizFansRedisKey = RedisKeyConstants.buildQuizFansKey(quizId);
                    // 获取该 Key 按照时间降序排序的前 3 个元素
                    operations.opsForZSet().reverseRange(quizFansRedisKey, 0, 2);
                }
                return null;
            }
        });

        // 构建接受问题粉丝 id map
        Map<Long, List<Long>> quizIdAndFansIdsMap = Maps.newHashMap();

        // 需要数据库查询的问题 id
        List<Long> needQueryQuizIds = Lists.newArrayList();

        // 遍历对应结果
        for (int i = 0; i < quizIds.size(); i++) {
            // 问题 ID
            Long quizId = quizIds.get(i);
            // 粉丝 IDS
            Set<Object> fansIdsSet = (Set<Object>) redisPipelineResult.get(i);

            // 如果不为空则添加至 Map 中，无需数据查询该问题关系
            if (CollUtil.isNotEmpty(fansIdsSet)) {
                // 转化类型
                List<Long> fansIds = fansIdsSet.stream().map(value -> Long.parseLong(value.toString())).toList();

                quizIdAndFansIdsMap.put(quizId, fansIds);
                continue;
            }

            // 需要数据库查询该问题关系
            needQueryQuizIds.add(quizId);
        }

        // 需要数据库查询的 quizIds 不为空
        if (CollUtil.isNotEmpty(needQueryQuizIds)) {
            // 通过窗口函数批量查询 quizId 的 top 10 的粉丝
            List<QuizFansDO> selectedTop10FansByQuizIds = quizFansDOMapper.selectTop10FansByQuizIds(needQueryQuizIds);

            // 如果查询结果不为空
            if (CollUtil.isNotEmpty(selectedTop10FansByQuizIds)) {

                // 按 quizId 分组
                Map<Long, List<QuizFansDO>> selectedTop10FansByQuizIdsMap = selectedTop10FansByQuizIds
                        .stream()
                        .collect(
                                Collectors.groupingBy(
                                        QuizFansDO::getQuizId,
                                        Collectors.mapping(
                                                value -> value,
                                                Collectors.toList()
                                        )
                                ));

                // 遍历添加到 quizIdAndFansIdsMap 中
                selectedTop10FansByQuizIdsMap.forEach((quizId, fans) -> {
                    // 取前三个粉丝 id
                    List<Long> fansList = fans.stream().map(QuizFansDO::getFansUserId).limit(3).toList();

                    quizIdAndFansIdsMap.put(quizId, fansList);
                });


                // 将问题关系异步同步到 redis
                threadPoolTaskExecutor.submit(() ->
                        syncQuizRelationZSetRedis(selectedTop10FansByQuizIdsMap));
            }
        }

        // 获取全部问题的粉丝集合大小
        int fansListSize = quizIdAndFansIdsMap.values().stream().mapToInt(List::size).sum();

        // 构建返回结果 Map
        Map<Long, List<String>> quizIdAndFansImagesMap = Maps.newHashMapWithExpectedSize(quizIdAndFansIdsMap.size());

        if (fansListSize > 0) {
            // 预分配数组空间 set 去重
            Set<Long> needQueryFansIdsSet = Sets.newLinkedHashSetWithExpectedSize(fansListSize);

            // 遍历 quizIdAndFansIdsMap 添加到 needQueryFansIds 中
            quizIdAndFansIdsMap.forEach((quizId, fansIds) -> needQueryFansIdsSet.addAll(fansIds));

            // 转化为 List
            List<Long> needQueryFansIdsList = needQueryFansIdsSet.stream().toList();

            // 调用用户 RPC
            List<FindUserByIdRspDTO> fansInfos = userRpcService.findUsersByIds(needQueryFansIdsList);

            // 构建粉丝 id 和用户头像 map
            Map<Long, String> fansIdAndImageMap = fansInfos.stream().collect(
                    Collectors.toMap(
                            FindUserByIdRspDTO::getId,
                            FindUserByIdRspDTO::getPhotoUrl,
                            // 键重复时保留 第一个键
                            (key1, key2) -> key1)
            );

            quizIdAndFansIdsMap.forEach((quizId, fansIds) -> {
                fansIds.forEach(fansId -> {
                    // 获取用户头像
                    String fansImage = fansIdAndImageMap.get(fansId);
                    // 添加到 fansImages 中, 当前 quizId 不存在时，创建一个空列表，存在时添加
                    quizIdAndFansImagesMap.computeIfAbsent(quizId, k -> Lists.newArrayList()).add(fansImage);
                });
            });
        }

        FindQuizFansTop3ByIdsRspDTO findQuizFansTop3ByIdsRspDTO = FindQuizFansTop3ByIdsRspDTO.builder()
                .quizIdAndFansImageMap(quizIdAndFansImagesMap)
                .build();

        return Response.success(findQuizFansTop3ByIdsRspDTO);
    }

    /**
     * 初始化 RBitmap
     *
     * @param quizFollowingDOS
     * @param expireSeconds
     * @param quizFollowingRedisKey
     */
    private void batchAddQuizFollowRBitmapAndExpire(List<QuizFollowingDO> quizFollowingDOS, long expireSeconds, String quizFollowingRedisKey) {
        // 构建 Lua 脚本参数
        List<Long> luaArgs = quizFollowingDOS.stream().map(QuizFollowingDO::getFollowingQuizId).collect(Collectors.toList());

        // 最后一位添加过期时间
        luaArgs.add(expireSeconds);

        log.info("==> 开始初始化 RBitmap, Redis Key: {}, Lua 脚本参数: {}", quizFollowingRedisKey, luaArgs);

        DefaultRedisScript<Long> script = new DefaultRedisScript<>();
        // lua 脚本路径
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/rbitmap_follow_batch_add_and_expire.lua")));
        // 返回值类型
        script.setResultType(Long.class);

        redisTemplate.execute(script, Collections.singletonList(quizFollowingRedisKey), luaArgs.toArray());
    }

    /**
     * 校验 Lua 脚本结果，根据状态码抛出对应的业务异常
     *
     * @param result
     */
    public void checkLuaScriptResult(Long result) {
        LuaResultEnum luaResultEnum = LuaResultEnum.valueOf(result);

        if (Objects.isNull(luaResultEnum)) {
            throw new RuntimeException("Lua 返回结果错误");
        }

        switch (luaResultEnum) {
            case ALREADY_FOLLOWED -> throw new BizException(ResponseCodeEnum.ALREADY_FOLLOWED);
        }
    }

    /**
     * 将问题关系异步同步到 redis
     *
     * @param quizIdAndQuizFansDOMap
     */
    private void syncQuizRelationZSetRedis(Map<Long, List<QuizFansDO>> quizIdAndQuizFansDOMap) {
        redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public Object execute(RedisOperations operations) {
                quizIdAndQuizFansDOMap.forEach((quizId, quizFansDOList) -> {

                    // 构建 Redis Key
                    String quizFansRedisKey = RedisKeyConstants.buildQuizFansKey(quizId);

                    // 遍历 quizFansDOList 添加到 ZSET 中
                    quizFansDOList.forEach(quizFansDO -> {
                        operations.opsForZSet().add(quizFansRedisKey,
                                quizFansDO.getFansUserId(),
                                DateUtils.localDateTime2Timestamp(quizFansDO.getCreateTime())
                        );
                    });

                    // 设置随机过期时间 (1 小时以内)
                    long expireTime = 60 * 30 + RandomUtil.randomInt(60 * 30);
                    operations.expire(quizFansRedisKey, expireTime, TimeUnit.SECONDS);
                });
                return null;
            }
        });
    }
}
